package com.lizelinteam.main3
import com.lizelinteam.utils.PathConfig
import org.apache.spark.sql.{SparkSession, functions}

object Fugou1 {
  def main(args: Array[String]): Unit = {
    // 创建 SparkSession，集群模式下不指定master
    val spark = SparkSession.builder()
      .appName("Repurchase Rate Calculation")
      .getOrCreate()


    // 读取数据，使用PathConfig管理输入路径
    val df = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(PathConfig.inputPath)

    // 按 user_id 和 item_id 分组并统计购买次数
    val grouped = df.groupBy("user_id", "item_id")
      .agg(functions.count("item_id").alias("purchase_count"))

    // 筛选购买次数大于 2 的商品
    val filtered = grouped.filter(functions.col("purchase_count") > 2)

    // 统计每个商品的复购次数
    val itemRepurchaseCount = filtered.groupBy("item_id")
      .agg(functions.sum("purchase_count").alias("total_repurchase_count"))

    // 按复购次数降序排序并取前十个
    val topTenItems = itemRepurchaseCount.orderBy(functions.desc("total_repurchase_count")).limit(10)


    // 输出到控制台
    println("购买商品复购次数前十的商品：")
    topTenItems.show()



    // 写入结果，使用PathConfig管理输出路径
    topTenItems.write
      .mode("overwrite")
      .option("header", "true")
      .csv(PathConfig.output41Path)



    println(s"结果已保存至：")
    println(s"商品复购次数排名：${PathConfig.output41Path}")


    // 停止 SparkSession
    spark.stop()
  }
}
